package com.intct.ods;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author gufg
 * @since 2025-10-21 10:49
 */
public class OdsVehicleInfoSQL {

    public static void main(String[] args) {
        // 获取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并度
        env.setParallelism(1);
        // 开启检查
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        // SQL环境
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 映射Mysql
        tenv.executeSql(" CREATE TABLE flink_vehicle_info(        " +
                "     id	                INT                            " +
                "     ,license              STRING                         " +
                "     ,series               STRING                         " +
                "     ,brand                STRING                         " +
                "     ,PRIMARY KEY(id) NOT ENFORCED                        " +
                " )WITH(                                                   " +
                "     'connector' = 'mysql-cdc'                            " +
                "     ,'hostname' = 'cdh-node'                             " +
                "     ,'port' = '13306'                                    " +
                "     ,'username' = 'root'                                 " +
                "     ,'password' = 'Test_090110'                          " +
                "     ,'scan.startup.mode' = 'initial'                     " +
                "     ,'database-name' = 'travel'                          " +
                "     ,'table-name' = 'vehicle_info'                       " +
                " )                                                        " );

        // 映射HBase
        tenv.executeSql(" CREATE TABLE hbase_vehicle_info(          " +
                "     id	STRING                                           " +
                "     ,info  ROW<                                            " +
                "               license	               STRING                " +
                "               ,series	               STRING                " +
                "               ,brand	               STRING                " +
                "     >                                                      " +
                " )WITH(                                                     " +
                "     'connector' = 'hbase-2.2'                              " +
                "     ,'zookeeper.quorum' = 'cdh-node:2181'                  " +
                "     ,'table-name' = 'dim:dim_vehicle_info'                 " +
                " )                                                          " );

        // 插入数据
        tenv.executeSql("INSERT INTO hbase_vehicle_info                " +
                        "SELECT                                                 " +
                        "       CAST(id AS STRING)                              " +
                        "       , ROW(                                          " +
                        "        license                                        " +
                        "       ,series                                         " +
                        "       ,brand                                          " +
                        "       ) as info                                       " +
                        "  FROM flink_vehicle_info                              " );
    }
}
